package com.aiwiown.snackmq.broker.handler;

import com.aiwiown.snackmq.broker.auth.AuthenticationHandler;
import com.aiwiown.snackmq.broker.auth.AuthorizationService;
import com.aiwiown.snackmq.broker.auth.JwtService;
import com.aiwiown.snackmq.broker.config.BrokerConfig;
import com.aiwiown.snackmq.broker.handler.console.BindQueueHandler;
import com.aiwiown.snackmq.broker.handler.console.CreateUserHandler;
import com.aiwiown.snackmq.broker.handler.console.DeleteExchangeHandler;
import com.aiwiown.snackmq.broker.handler.console.DeleteRoleHandler;
import com.aiwiown.snackmq.broker.handler.console.DeleteTopicHandler;
import com.aiwiown.snackmq.broker.handler.console.DeleteUserHandler;
import com.aiwiown.snackmq.broker.handler.console.DiagnosePartitionHandler;
import com.aiwiown.snackmq.broker.handler.console.GetClusterOverviewHandler;
import com.aiwiown.snackmq.broker.handler.console.ListBrokersHandler;
import com.aiwiown.snackmq.broker.handler.console.ListConsumerGroupsHandler;
import com.aiwiown.snackmq.broker.handler.console.ListExchangesHandler;
import com.aiwiown.snackmq.broker.handler.console.ListRolesHandler;
import com.aiwiown.snackmq.broker.handler.console.ListTopicsHandler;
import com.aiwiown.snackmq.broker.handler.console.ListUsersHandler;
import com.aiwiown.snackmq.broker.handler.console.ManualRepairPartitionHandler;
import com.aiwiown.snackmq.broker.handler.console.RepairPartitionHandler;
import com.aiwiown.snackmq.broker.handler.console.SetRolePermissionsHandler;
import com.aiwiown.snackmq.broker.handler.console.UnbindQueueHandler;
import com.aiwiown.snackmq.cluster.metadata.MetadataManager;
import com.aiwiown.snackmq.cluster.raft.SnackMQStateMachine;
import com.aiwiown.snackmq.cluster.raft.manager.SnackMQRaftNodeManager;
import com.aiwiown.snackmq.common.cluster.BrokerId;
import com.aiwiown.snackmq.common.handler.MessageHandler;
import com.aiwiown.snackmq.management.BrokerMetrics;
import com.aiwiown.snackmq.management.scaling.AutoScalerService;
import com.aiwiown.snackmq.network.connection.ConnectionManager;
import com.aiwiown.snackmq.storage.StorageService;
import com.aiwiown.snackmq.storage.service.PartitionedStorageService;
import com.aiwiown.snackmq.storage.service.ZeroCopyStorageService;
import com.aiwiown.snackmq.storage.transaction.TransactionLog;

import java.util.ArrayList;
import java.util.List;

/**
 * 消息处理器工厂，负责创建和组装所有业务逻辑处理器。
 */
public class MessageHandlerFactory {

    /**
     * 创建并组装所有需要的业务逻辑处理器，支持零拷贝存储服务。
     *
     * @param metadataManager        用于管理集群元数据的管理器。
     * @param dataRaftNodeManager    用于管理数据 Raft 节点的管理器。
     * @param storageService         存储服务，用于持久化消息。
     * @param zeroCopyStorageService 零拷贝存储服务，用于高性能读取。
     * @param stateMachine           数据状态机，用于处理状态转换。
     * @param brokerMetrics          Broker 性能监控器。
     * @param autoScalerService      自动扩展服务。
     * @return 一个包含所有业务逻辑处理器的列表。
     */
    public static List<MessageHandler> createHandlers(
            MetadataManager metadataManager,
            SnackMQRaftNodeManager dataRaftNodeManager,
            StorageService storageService,
            ZeroCopyStorageService zeroCopyStorageService,
            SnackMQStateMachine stateMachine,
            BrokerMetrics brokerMetrics,
            AutoScalerService autoScalerService,
            TransactionLog transactionLog, // 【修改】增加 BrokerId 依赖
            BrokerConfig brokerConfig, BrokerId selfId) {

        List<MessageHandler> handlers = new ArrayList<>();
        // --- 【新增】初始化认证和授权服务 ---
        JwtService jwtService = new JwtService(brokerConfig.getJwtSecretKey(), brokerConfig.getJwtExpirationMs());
        AuthorizationService authorizationService = new AuthorizationService(metadataManager, jwtService, brokerConfig.isAuthEnabled());
        // --- 【新增】创建并注册认证处理器 ---
        handlers.add(new AuthenticationHandler(metadataManager, jwtService));

        // --- 关键修复：将数据状态机 (stateMachine) 注入到 PublishMessageHandler ---
        handlers.add(new PublishMessageHandler(
                dataRaftNodeManager,
                metadataManager,
                brokerMetrics.getMessageMetricEmitter(),
                autoScalerService,
                stateMachine, // <-- 注入 stateMachine
                authorizationService
        ));
        // --- 【新增】注册 RegisterProducerHandler ---
        handlers.add(new RegisterProducerHandler(ConnectionManager.getInstance()));
        // --- 【新增】注册事务消息处理器 ---
        handlers.add(new PrepareMessageHandler(transactionLog, authorizationService));
        // --- 【新增】注册事务二次确认处理器 ---
        handlers.add(new CommitMessageHandler(transactionLog, dataRaftNodeManager, authorizationService, stateMachine));
        handlers.add(new RollbackMessageHandler(transactionLog, authorizationService));

        // 以下处理器负责处理各种元数据操作
        handlers.add(new CreateTopicHandler(metadataManager, authorizationService));
        handlers.add(new DeclareExchangeHandler(metadataManager, authorizationService));
        handlers.add(new DeleteTopicHandler(metadataManager, dataRaftNodeManager, authorizationService));
        handlers.add(new BindQueueHandler(metadataManager, authorizationService));
        handlers.add(new UnbindQueueHandler(metadataManager, authorizationService)); // 【新增】
        handlers.add(new DeleteExchangeHandler(metadataManager, authorizationService));

        // 提交偏移量的处理器，依赖数据 Raft 节点
        handlers.add(new AckMessageHandler(dataRaftNodeManager, authorizationService, brokerMetrics.getMessageMetricEmitter()));

        // 获取偏移量的处理器，直接从数据状态机读取
        handlers.add(new FetchOffsetHandler(stateMachine, metadataManager));
        handlers.add(new FetchTopicMetadataHandler(metadataManager));
        // 消费处理器直接从存储服务读取数据
        if (storageService instanceof PartitionedStorageService) {
            PartitionedStorageService pss = (PartitionedStorageService) storageService;
            handlers.add(new ConsumeMessageHandler(
                    pss,
                    zeroCopyStorageService, // 可以为 null
                    brokerConfig.isZeroCopyEnabled(),
                    authorizationService,
                    metadataManager, // <-- 【新增】注入 MetadataManager
                    selfId           // <-- 【新增】注入 BrokerId
            ));
            // 【修复】将 PullMessageHandler 的初始化移到此处，并使用正确的参数
            handlers.add(new PullMessageHandler(pss, authorizationService));
        } else {
            // 【修复】更新异常信息
            throw new IllegalArgumentException("ConsumeMessageHandler and PullMessageHandler require a PartitionedStorageService instance.");
        }
        // --- 【修复】删除下面这个重复的代码块 ---
        // if (storageService instanceof PartitionedStorageService) { ... }

        // --- 【新增】注册控制台管理处理器 ---
        handlers.add(new GetClusterOverviewHandler(metadataManager, brokerMetrics, authorizationService, stateMachine));
        handlers.add(new ListTopicsHandler(metadataManager, authorizationService));
        handlers.add(new ListConsumerGroupsHandler(stateMachine, dataRaftNodeManager, authorizationService));
        handlers.add(new ListBrokersHandler(metadataManager, dataRaftNodeManager, authorizationService));
        handlers.add(new ListUsersHandler(metadataManager, authorizationService));
        handlers.add(new ListRolesHandler(metadataManager, authorizationService));
        handlers.add(new CreateUserHandler(metadataManager, authorizationService));
        handlers.add(new SetRolePermissionsHandler(metadataManager, authorizationService));
        handlers.add(new DeleteUserHandler(metadataManager, authorizationService));
        handlers.add(new DeleteRoleHandler(metadataManager, authorizationService));
        handlers.add(new ListExchangesHandler(metadataManager, authorizationService));
        handlers.add(new DiagnosePartitionHandler(storageService, authorizationService));
        handlers.add(new RepairPartitionHandler(storageService, authorizationService));
        handlers.add(new ManualRepairPartitionHandler(storageService, authorizationService));

        return handlers;
    }
}
